package Java7_Concurrency_Cookbook.concurrent_custom.custom_priority_transfer_queue;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author lixiangke
 * @date 2017/11/28
 */
public class MyPriorityTransferQueue<E> extends PriorityBlockingQueue<E>
        implements TransferQueue<E> {

    /**
     * 等待从数据结构中读取元素的消费者数量
     */
    private AtomicInteger counter;

    /**
     * 存储已传递(尚未被消费)的元素
     */
    private LinkedBlockingQueue<E> transfered;

    /**
     * 用来控制生产者, 消费者, 实现仅有一个线程操作数据结构的目标
     */
    private ReentrantLock lock;

    public MyPriorityTransferQueue() {
        this.counter = new AtomicInteger(0);
        this.lock = new ReentrantLock();
        this.transfered = new LinkedBlockingQueue<>();
    }

    /**
     * 该方法尝试立即将元素发送到一个正在等待的消费者. <br/>
     * 如果没有等待中的消费者, 该方法返回false;
     */
    @Override
    public boolean tryTransfer(E e) {
        lock.lock();
        boolean value;
        if (counter.get() == 0) {
            value = false;
        } else {
            put(e);
            value = true;
        }
        lock.unlock();
        return value;
    }

    /**
     * 该方法尝试立即将元素发送到一个正在等待的消费者. <br/>
     * 如果没有等待中的消费者, 该方法将元素存储到transfered队列,
     * 并等待出现试图获得元素的第一个消费者, 在这之前线程将被阻塞<br/>
     * @param e
     * @throws InterruptedException
     */
    @Override
    public void transfer(E e) throws InterruptedException {
        lock.lock();
        if (counter.get() != 0) {
            put(e);
            lock.unlock();
        } else {
            transfered.add(e);
            lock.unlock();
            synchronized (e) {
                e.wait();
            }
        }
    }

    /**
     * 限定时间版本的tryTransfer
     */
    @Override
    public boolean tryTransfer(E e, long timeout, TimeUnit unit) throws InterruptedException {
        lock.lock();
        if (counter.get() != 0) {
            put(e);
            lock.unlock();
            return true;
        } else {
            transfered.add(e);
            long newTimeout = TimeUnit.MILLISECONDS.convert(timeout, unit);
            lock.unlock();
            e.wait(newTimeout);
            lock.lock();
            if (transfered.contains(e)) {
                transfered.remove(e);
                lock.unlock();
                return false;
            } else {
                lock.unlock();
                return true;
            }
        }
    }

    @Override
    public boolean hasWaitingConsumer() {
        return counter.get() != 0;
    }

    @Override
    public int getWaitingConsumerCount() {
        return counter.get();
    }

    @Override
    public E take() throws InterruptedException {
        lock.lock();
        counter.incrementAndGet();
        E value = transfered.poll();
        if (value == null) {
            lock.unlock();
            value = super.take();
            lock.lock();
        } else {
            synchronized (value) {
                value.notify();
            }
        }
        counter.decrementAndGet();
        lock.unlock();
        return value;
    }
}
